﻿/* Copyright (C) 2007-2010 STS Soft

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections;
using STSdb.General.Buffers;
using STSdb.General.Collections;
using STSdb.Persist;
using STSdb.Data;
using STSdb.Files;
using System.Diagnostics;

namespace STSdb.Plan
{
    public class StressBuffer<TKey, TRecord> : IEnumerable<Row<TKey, TRecord>>
    {
        private RawFile TmpFile;
        private List<XTable<TKey, TRecord>> tables = new List<XTable<TKey, TRecord>>();
        private XTable<TKey, TRecord> CurrTable;
        private XTable<TKey, bool> RangeTable;

        public StorageEngine Repository { get; private set; }
        public uint PartitionCapacity { get; set; }
        public ulong RangeCapacity { get; set; }
        public IKeyMap<TKey> KeyMap { get; private set; }
        public IBinaryPersist<IIndexer<TRecord>> RecordPersist { get; private set; }

        public StressBuffer(StorageEngine repository)
            : this(repository, null, null)
        {
        }

        public StressBuffer(StorageEngine repository, IKeyMap<TKey> keyMap, IBinaryPersist<IIndexer<TRecord>> recordPersist)
        {
            if (repository == null)
                throw new ArgumentNullException("repository");

            Repository = repository;
            KeyMap = keyMap;
            RecordPersist = recordPersist;

            TmpFile = Repository.FileSystem.CreateFile();
            RangeCapacity = 1000;
            PartitionCapacity = 1000000;

            RangeTable = new XTable<TKey, bool>(TmpFile, TmpFile.Length);
            if (KeyMap != null)
                RangeTable.KeyMap = KeyMap;
            RangeTable.CacheCapacity = uint.MaxValue;

            CurrTable = AddTable();
        }

        private XTable<TKey, TRecord> AddTable()
        {
            XTable<TKey, TRecord> table = new XTable<TKey, TRecord>(TmpFile, TmpFile.Length);
            table.CacheCapacity = uint.MaxValue;
            if (KeyMap != null)
                table.KeyMap = KeyMap;
            if (RecordPersist != null)
                table.RecordPersist = RecordPersist;
            tables.Add(table);

            return table;
        }

        private void FillRange(XTable<TKey, TRecord> table)
        {
            uint count = 0;
            foreach (var row in table.Forward())
            {
                if (count % RangeCapacity == 0)
                    RangeTable[row.Key] = true;

                count++;
            }
        }

        public void Add(TKey key, TRecord record)
        {
            if (CurrTable.Count >= PartitionCapacity)
            {
                FillRange(CurrTable);
                CurrTable.CacheCapacity = 1;

                CurrTable = AddTable();
            }

            CurrTable[key] = record;
        }

        #region IEnumerable<Row<TKey,TRecord>> Members

        public IEnumerator<Row<TKey, TRecord>> GetEnumerator()
        {
            try
            {
                if (tables.Count == 0)
                    yield break;

                if (tables.Count == 1)
                {
                    foreach (var row in tables[0].Forward())
                        yield return row;

                    yield break;
                }

                FillRange(CurrTable);
                CurrTable.CacheCapacity = 1;

                XTable<TKey, TRecord> tmp = new XTable<TKey, TRecord>(TmpFile, TmpFile.Length);
                if (KeyMap != null)
                    tmp.KeyMap = KeyMap;
                tmp.CacheCapacity = uint.MaxValue;

                List<TKey> keys = new List<TKey>((int)RangeTable.Count);
                ulong count = 0;
                foreach (var row in RangeTable.Forward())
                {
                    if (keys.Count == 0)
                        keys.Add(row.Key);

                    count += RangeCapacity;
                    if (count >= PartitionCapacity)
                    {
                        keys.Add(row.Key);
                        count = 0;
                    }
                }
                RangeTable.Close();
                RangeTable = null;

                for (int i = 0; i < keys.Count; i++)
                {
                    TKey left = keys[i];
                    byte[] leftIndex = tmp.KeyMap.Direct(left);
                    if (i > 0)
                    {
                        if (leftIndex.Increment())
                            leftIndex = new byte[leftIndex.Length + 1];
                    }
                    if (i + 1 < keys.Count)
                    {
                        TKey right = keys[i + 1];
                        byte[] rightIndex = tmp.KeyMap.Direct(right);
                        foreach (var node in tables)
                        {
                            foreach (var row in node.Forward(leftIndex, rightIndex))
                                tmp[row.Key] = row.Record;
                        }
                    }
                    else
                    {
                        foreach (var node in tables)
                        {
                            foreach (var row in node.Forward(leftIndex))
                                tmp[row.Key] = row.Record;
                        }
                    }

                    foreach (var row in tmp.Forward())
                        yield return row;

                    tmp.Clear();
                }

                foreach (var table in tables)
                    table.Close();
            }
            finally
            {
                TmpFile.Close();
                Repository.FileSystem.DeleteFile(TmpFile.Handle);
            }
        }

        #endregion

        #region IEnumerable Members

        IEnumerator IEnumerable.GetEnumerator()
        {
            return GetEnumerator();
        }

        #endregion
    }
}
